package com.hucais.agg.service

import com.hucais.agg.bean.{DwtCopyrightsInfo, DwtCopyrightsTopn}
import com.hucais.agg.dao.SourceIndexDataDao
import com.hucais.core.utils.DefaultPropertiesUtil
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.elasticsearch.spark.sql.EsSparkSQL

object CopyrightsTopNService {
  def action(ssc: SparkContext, sparkSession: SparkSession): Unit = {
    // 获取book_base_info数据
    val bookBaseInfoDS = SourceIndexDataDao.getSearchInfoData(sparkSession)
    bookBaseInfoDS.cache()
    bookBaseInfoDS.createOrReplaceTempView("dwd_book_base_info")
    // 获取上架产品数据
    val shelfProductDS = SourceIndexDataDao.getShelfProduct(sparkSession)
    shelfProductDS.cache()
    shelfProductDS.createOrReplaceTempView("ods_shelf_products")
    // 获取产品动销情况数据
    val productSalesDS = SourceIndexDataDao.getProductSales(sparkSession)
    productSalesDS.cache()
    productSalesDS.createOrReplaceTempView("ods_product_sales_info")

    // 输出统计数据
    outPutTopN(sparkSession)
    outPutInfo(sparkSession)
  }

  private def outPutTopN(sparkSession: SparkSession): Unit = {
    import sparkSession.implicits._
    val resultDS = sparkSession.sql(
      s"""
         |select
         |	copyright_person,coalesce (year,'全部') as year,coalesce (month,'全部') as month,
         |	sum(copyright_num) as copyright_num,sum(shelf_num) as shelf_num,date_format(date_add(current_date,-1),'yyyy-MM-dd') as create_time
         |from
         |(
         |	select
         |		uploader as copyright_person,SUBSTR(SUBSTR(copyright_month,1,7),1,4) as year,SUBSTR(SUBSTR(copyright_month,1,7),6,2) as month,
         |		count(*) as copyright_num,0 as shelf_num
         |	from dwd_book_base_info where uploader is not null
         |	group by uploader,SUBSTR(copyright_month,1,7)
         |	union all
         |	select
         |		copyright_person,SUBSTR(SUBSTR(shelf_time,1,7),1,4) as year,SUBSTR(SUBSTR(shelf_time,1,7),6,2) as month,
         |		0 as copyright_num,count(*) as shelf_num
         |	from ods_shelf_products
         |	group by copyright_person,SUBSTR(shelf_time,1,7)
         |)a
         |group by copyright_person,year,month
         |grouping sets(
         |	(copyright_person),(copyright_person,year),(copyright_person,month),(copyright_person,year,month)
         |)
         |""".stripMargin).as[DwtCopyrightsTopn]
    EsSparkSQL.saveToEs(resultDS, DefaultPropertiesUtil.get("index.copyrights.topn"))
  }

  private def outPutInfo(sparkSession: SparkSession): Unit = {
    import sparkSession.implicits._
    val tmpDF = sparkSession.sql(
      """
        |select
        |		coalesce (year,'全部') as year,coalesce (month,'全部') as month,
        |		sum(a.total_copyright_num) as total_copyright_num,sum(a.total_shelf_num) as total_shelf_num,sum(a.total_sales) as total_sales
        |	from
        |	(
        |		select
        |			SUBSTR(SUBSTR(copyright_month,1,7),1,4) as year,SUBSTR(SUBSTR(copyright_month,1,7),6,2) as month,
        |			count(*) as total_copyright_num,0 as total_shelf_num,0 as total_sales
        |		from dwd_book_base_info where copyright_month is not null
        |		group by SUBSTR(copyright_month,1,7)
        |		union all
        |		select
        |			SUBSTR(SUBSTR(shelf_time,1,7),1,4) as year,SUBSTR(SUBSTR(shelf_time,1,7),6,2) as month,
        |			0 as total_copyright_num,count(*) as total_shelf_num,0 as total_sales
        |		from ods_shelf_products
        |		group by SUBSTR(shelf_time,1,7)
        |		union all
        |		select
        |			SUBSTR(SUBSTR(statistical_month,1,7),1,4) as year,SUBSTR(SUBSTR(statistical_month,1,7),6,2) as month,
        |			0 as total_copyright_num,0 as total_shelf_num,sum(sales) as total_sales
        |		from ods_product_sales_info
        |		group by SUBSTR(statistical_month,1,7)
        |	)a
        |	group by year,month
        |	grouping sets((year),(month),(year,month),(month,year))
        |""".stripMargin)
    tmpDF.createOrReplaceTempView("tmp_info")

    val resultDS = sparkSession.sql(
      s"""
         |select year,month,total_copyright_num,total_shelf_num,total_sales,date_format(date_add(current_date,-1),'yyyy-MM-dd') as create_time
         |from tmp_info
         |union all
         |select
         |	year,'全部' as month,
         |	sum(total_copyright_num) as total_copyright_num,sum(total_shelf_num) as total_shelf_num,sum(total_sales) as total_sales,
         |	date_format(date_add(current_date,-1),'yyyy-MM-dd') as create_time
         |from tmp_info where year='全部' group by year
         |""".stripMargin).as[DwtCopyrightsInfo]
    EsSparkSQL.saveToEs(resultDS, DefaultPropertiesUtil.get("index.copyrights.info"))
  }

}
